package mr

import (
	"fmt"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"sync"
	"time"
)

type mapjob_guard struct {
	map_file string
	status int // 0: not shedule, 1:sheduled, 2:done
	mapJobStartTime time.Time
	mu sync.Mutex
}

type redjob_guard struct {
	reduce_file []string
	status int // 0: not shedule, 1:sheduled, 2:done
	JobStartTime time.Time
	mu sync.Mutex
}

type Coordinator struct {
	// Your definitions here.
	nReduce int
	mapJobs []mapjob_guard
	reduceJobs []redjob_guard
	mapDone bool
	reduceDone bool
}

// Your code here -- RPC handlers for the worker to call.

//
// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
//
func (c *Coordinator) AssnJob(args *AssnJobArgs, reply *AssnJobReply) error {
	reply.JobIndex=-1
	if !c.mapDone {
		i := 0
		for ; i < len(c.mapJobs); i++ {
			c.mapJobs[i].mu.Lock()
			if c.mapJobs[i].status==0 {
				c.mapJobs[i].status=1
				reply.Inputfiles=append(reply.Inputfiles, c.mapJobs[i].map_file)
				reply.IsMapJob=true
				reply.JobIndex=i
				reply.Nreduce=c.nReduce
				c.mapJobs[i].mapJobStartTime=time.Now()
				c.mapJobs[i].mu.Unlock()
				break
			} else if c.mapJobs[i].status==1 && time.Since(c.mapJobs[i].mapJobStartTime) > 10*time.Second {
				reply.Inputfiles=append(reply.Inputfiles, c.mapJobs[i].map_file)
				reply.IsMapJob=true
				reply.JobIndex=i
				reply.Nreduce=c.nReduce
				c.mapJobs[i].mapJobStartTime=time.Now()
				c.mapJobs[i].mu.Unlock()
				break
			}
			c.mapJobs[i].mu.Unlock()
			if c.mapDone {
				break;
			}
		}
		if !c.mapDone && i == len(c.mapJobs) {
			j := 0
			for ; j < len(c.mapJobs); j++ {
				c.mapJobs[j].mu.Lock()
				if c.mapJobs[j].status != 2 {
					c.mapJobs[j].mu.Unlock()
					break
				}
				c.mapJobs[j].mu.Unlock()
			}
			if j == len(c.mapJobs) {
				c.mapDone=true
			}
		}
	}
	
	if c.mapDone && !c.reduceDone {
		i := 0
		for ; i < c.nReduce; i++ {
			c.reduceJobs[i].mu.Lock()
			if c.reduceJobs[i].status==0 {
				c.reduceJobs[i].status=1
				reply.Inputfiles = append(reply.Inputfiles, c.reduceJobs[i].reduce_file...)
				reply.IsMapJob=false
				reply.JobIndex = i
				c.reduceJobs[i].JobStartTime=time.Now()
				c.reduceJobs[i].mu.Unlock()
				break
			} else if c.reduceJobs[i].status==1 && time.Since(c.reduceJobs[i].JobStartTime) > 10*time.Second {
				reply.Inputfiles = append(reply.Inputfiles, c.reduceJobs[i].reduce_file...)
				reply.IsMapJob=false
				reply.JobIndex=i
				c.reduceJobs[i].JobStartTime=time.Now()
				c.reduceJobs[i].mu.Unlock()
				break
			}
			c.reduceJobs[i].mu.Unlock()
			if c.reduceDone {
				break;
			}
		}
		if !c.reduceDone && i == c.nReduce {
			j := 0
			for ; j < c.nReduce; j++ {
				c.reduceJobs[j].mu.Lock()
				if c.reduceJobs[j].status != 2 {
					c.reduceJobs[j].mu.Unlock()
					break
				}
				c.reduceJobs[j].mu.Unlock()
			}
			if j == c.nReduce {
				c.reduceDone = true
			}
		}
	}

	return nil
}

func (c *Coordinator) FinishJob(args *FinishJobArgs, reply *FinishJobReply) error {
	if args.IsMapJob {
		c.mapJobs[args.JobIndex].mu.Lock()
		c.mapJobs[args.JobIndex].status=2
		c.mapJobs[args.JobIndex].mu.Unlock()
	} else {
		c.reduceJobs[args.JobIndex].mu.Lock()
		c.reduceJobs[args.JobIndex].status=2
		c.reduceJobs[args.JobIndex].mu.Unlock()
	}
	return nil
}

//
// start a thread that listens for RPCs from worker.go
//
func (c *Coordinator) server() {
	rpc.Register(c)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := coordinatorSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
//
func (c *Coordinator) Done() bool {
	return c.reduceDone
}

//
// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{}

	// Your code here.
	c.mapJobs=make([]mapjob_guard, len(files))
	c.reduceJobs=make([]redjob_guard, nReduce)
	for i:=0; i<len(c.mapJobs); i++ {
		c.mapJobs[i].map_file=files[i]
	}
	for i:=0; i<nReduce; i++ {
		for j:=0; j<len(c.mapJobs); j++ {
			c.reduceJobs[i].reduce_file = append(c.reduceJobs[i].reduce_file, fmt.Sprintf("mr-%d-%d", j, i))
		}
	}
	c.nReduce=nReduce
	c.server()
	return &c
}
